package com.kedacom.demo.spring.websocket.rpc.client;

import java.util.UUID;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import com.kedacom.demo.spring.websocket.WebSocketClient;
import com.kedacom.demo.spring.websocket.rpc.PendingRequest;
import com.kedacom.demo.spring.websocket.rpc.PendingRequestHolder;
import com.kedacom.demo.spring.websocket.rpc.Request;
import com.kedacom.demo.spring.websocket.rpc.Response;
import com.kedacom.demo.spring.websocket.utils.JacksonUtils;

/**
 * RPCRequestSender
 * @author luocanfeng
 * @date 2017年8月11日
 */
// @Service
public final class RPCWebSocketClient extends WebSocketClient {

	private Logger logger = LoggerFactory.getLogger(getClass());

	private int timeout = 30;

	@Resource
	private PendingRequestHolder pendingRequestHolder;

	public RPCWebSocketClient() {
		super();
	}

	public RPCWebSocketClient(String url) {
		super(url);
	}

	public RPCWebSocketClient(String url, PendingRequestHolder pendingRequestHolder) {
		super(url);
		this.pendingRequestHolder = pendingRequestHolder;
	}

	@Override
	protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
		super.handleTextMessage(session, message);

		// TODO: 处理收到消息后的业务逻辑
		String text = message.getPayload();
		Response response = JacksonUtils.fromJsonString(text, Response.class);
		if (response == null) {
			logger.error("receive message with wrong format: {}", text);
			return;
		}
		String requestId = response.getId();
		if (StringUtils.isEmpty(requestId)) {
			// TODO: 处理推送消息
			logger.debug("It's not a response message.");
		} else {
			PendingRequest pendingRequest = pendingRequestHolder.get(requestId);
			pendingRequest.getQueue().add(text);
		}
	}

	public Response request(String type, String requestData) {
		String requestId = UUID.randomUUID().toString();
		PendingRequest pendingRequest = new PendingRequest(requestId);
		pendingRequestHolder.put(requestId, pendingRequest);

		Request request = new Request(requestId, type, requestData);
		this.sendMessage(JacksonUtils.toJsonString(request));
		return this.waitForResponse(pendingRequest);
	}

	private Response waitForResponse(PendingRequest pendingRequest) {
		LinkedBlockingQueue<String> replyHandoff = pendingRequest.getQueue();
		try {
			// receive the response
			String response = (timeout < 0) ? replyHandoff.take() : replyHandoff.poll(timeout, TimeUnit.SECONDS);
			if (response == null) {
				// timeout
				throw new RuntimeException(
						"Timeout while waiting for response of request with id: " + pendingRequest.getRequestId());
			}

			logger.debug("receive response '{}' for request with id: ", response, pendingRequest.getRequestId());

			// remove the pending request from pending request holder
			pendingRequestHolder.remove(pendingRequest.getRequestId());

			return JacksonUtils.fromJsonString(response, Response.class);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	public void setPendingRequestHolder(PendingRequestHolder pendingRequestHolder) {
		this.pendingRequestHolder = pendingRequestHolder;
	}

}
